package com.ververica.cdc.guass.sink.jdbc.mysql.testutil;

import com.ververica.cdc.guass.sink.jdbc.testutils.DatabaseExtension;
import com.ververica.cdc.guass.sink.jdbc.testutils.DatabaseMetadata;
import com.ververica.cdc.guass.sink.jdbc.testutils.DatabaseResource;
import org.apache.flink.util.FlinkRuntimeException;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;

public class MySqlDatabase extends DatabaseExtension {

    private static final MySqlMetadata metadata = new MySqlMetadata();

    public static MySqlMetadata getMetadata() {
        return metadata;
    }

    @Override
    protected DatabaseMetadata getMetadataDB() {
        return getMetadata();
    }

    @Override
    protected DatabaseResource getResource() {
        return new DatabaseResource() {
            @Override
            public void start() {
                try {
                    Class.forName(metadata.getDriverClass());
                    try (Connection conn = DriverManager.getConnection(
                            metadata.getJdbcUrlWithCredentials());
                         Statement stmt = conn.createStatement()) {
                        // 创建测试数据库
                        stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS " + metadata.getDbName());

                        // 切换到创建的数据库
                        stmt.executeUpdate("USE " + metadata.getDbName());

                        stmt.executeUpdate("drop table IF EXISTS" + metadata.getDbName()+ ".dl_job_history");

                    }
                } catch (Exception e) {
                    throw new FlinkRuntimeException(e);
                }
            }

            @Override
            public void stop() {

            }

        };
    }
}